package com.atguigu.flink.watermark;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2023/2/28
 *
 *  水印只在以下场景有用:
 *          基于EventTime的窗口运算
 *          定时器
 *          CEP
 */
public class Demo2_EventTimeWindow
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.disableOperatorChaining();
        //设置水印周期发送的间隔
        env.getConfig().setAutoWatermarkInterval(2000);


        //构造水印的制作策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            //选择策略
            .<WaterSensor>forMonotonousTimestamps()
            //告诉策略，数据的哪个属性是 时间属性
            .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>()
            {

                @Override
                public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                    //返回毫秒时间戳
                    return element.getTs();
                }
            });

        //刚上手，并行度必须是1
        env.setParallelism(1);


        /*
                以5s的滚动的keyed窗口为例
         */
        env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId)
            /*
                    窗口的范围： [0,4999], [5000,9999)
                            数据进入哪个窗口取决于 数据的事件时间属性 是否在 窗口的范围内。

                    水印:   只是时间作用，当系统的时间到了窗口的 end时，就会触发窗口的运算。
             */
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>()
            {
                @Override
                public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                    List<WaterSensor> waterSensors = MyUtil.parseToList(elements);
                    out.collect(key + ":" + context.window() + " :" + waterSensors);
                }
            })
            .print();


            env.execute();

    }
}
